package com.gitee.ywj1352.kafka;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import javax.sound.midi.Receiver;

public class ActiveMqListener implements MessageListener {
    //默认代理地址 "failover://tcp://localhost:61616"  服务器地址不同IP修改不同的IP
    private static final String BROKER_URL= ActiveMQConnection.DEFAULT_BROKER_URL;
    //消息队列名称
    private static final String SUBJECT="YWJ-TEST-QUEUE";
    public static void main(String[] args) throws JMSException {
        //初始化连接工厂
        ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(BROKER_URL);
        //建立连接
        Connection conn= connectionFactory.createConnection();
        //启动连接
        conn.start();
        //创建Session，此方法第一个参数表示会话是否在事务中执行，第二个参数设定会话的应答模式
        Session session= conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //创建目标队列
        useTopic(session);
        useTopic(session);
    }

    private static void useQueue(Session session) throws JMSException {
        Destination dest= session.createQueue(SUBJECT);
        //通过session创建消息的接收者
        MessageConsumer consumer= session.createConsumer(dest);
        //初始化监听
        ActiveMqListener receiver=new ActiveMqListener();
        //给接收者添加监听对象
        consumer.setMessageListener(receiver);
    }

    private static void useTopic(Session session) throws JMSException {
        Topic dest= session.createTopic("YWJ-TEST-TOPIC");
        //通过session创建消息的接收者
        MessageConsumer consumer= session.createConsumer(dest);
        //初始化监听
        ActiveMqListener receiver=new ActiveMqListener();
        //给接收者添加监听对象
        consumer.setMessageListener(receiver);
    }

    public void onMessage(Message arg0) {
        TextMessage message=(TextMessage) arg0;
        try {
            System.out.println(message.getText());
            //Thread.sleep(4000);
        } catch (Exception e) {
        }

    }
}